package PassengerVehicle

import java.io.{BufferedReader, FileReader}
import java.util.Properties
import PassengerVehicle.ylpzs.readFile
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, SparkSession}

object SendChuFactors {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.WARN)
	val cdict = readFile("/Users/chuxiaowei/IdeaProjects/两客一货/dataprocess/src/main/resources/incity.txt")
    val pdict = readFile("/Users/chuxiaowei/IdeaProjects/两客一货/dataprocess/src/main/resources/inprovince.txt")
    //创建sparkSession
    val spark: SparkSession = SparkSession.builder().appName("FactorsInOut").master("local[4]").config("spark.debug.maxToStringFields", "1000").getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    //创建properties对象 设置连接mysql的信息
    val prop: Properties = new Properties()
    prop.setProperty("user", "root")
    prop.setProperty("password", "123456")
    //读取mysql数据
    val mysqlDF: DataFrame = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/2k1w?useUnicode=true&characterEncoding=UTF-8", "ticket_smz3706", prop)
    val vehicleDF: DataFrame = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/2k1w?useUnicode=true&characterEncoding=UTF-8", "zhjg_zy_vehicle", prop)
    import spark.sql
    mysqlDF.filter("SCKCCPHM ='' or SCKCCPHM like '鲁F%' or SCKCCPHM like '鲁Y%'").createOrReplaceTempView("passenger")
    vehicleDF.filter("VEH_STATUS = 10").createOrReplaceTempView("vehicle")
    val dataframe = spark.sql("select XL,FCRQSJ,SCKCCPHM,DDZ from passenger")
    dataframe.filter(row => {
      val xl = row.getString(0)
      val ddz = row.getString(3)
      matchDDZ(xl, ddz, cdict)
    }).createOrReplaceTempView("a")


    sql("select SCKCCPHM,XL,FCRQSJ,first(DDZ) as DDZ from a GROUP BY XL,SCKCCPHM,FCRQSJ").createOrReplaceTempView("b")
    sql("select b.XL, b.SCKCCPHM,first(b.DDZ) as DDZ,count(1) as checi,if(b.SCKCCPHM = '',50,b.SCKCCPHM) as seat from b group by XL,SCKCCPHM").createOrReplaceTempView("c")
    sql("select a.XL, a.SCKCCPHM,first(a.DDZ) as DDZ,first(a.checi) as checi,if(a.SCKCCPHM='' or a.SCKCCPHM is null,50,first(b.SEAT_COUNT)) as seat from c a left join vehicle b on a.SCKCCPHM = b.VEH_NO group by XL,SCKCCPHM").createOrReplaceTempView("sss")
    import spark.implicits._
    sql("select XL,SCKCCPHM,DDZ,checi,seat,if(seat is null,0,seat*checi) as yl from sss").groupBy("XL").sum("yl").withColumn("yl", $"sum(yl)")
      .createOrReplaceTempView("d")
    sql("select XL from a").groupBy("XL").count().createOrReplaceTempView("e")
    sql("select d.XL,d.yl,e.count,e.count/yl as factors from d left join e on d.XL=e.XL").collect().foreach(r => {
      println(r.getString(0) + ",载客率:" + r.getDecimal(3))
    })

    /*    val outcityValue = dataframe.rdd.filter(row => {
          val xl = row.getString(0)
          val ddz = row.getString(4)
          !matchDDZ(xl, ddz, cdict) && matchDDZ(xl, ddz, pdict)
        }).count()*/

    /*    val outprovinceValue = dataframe.rdd.filter(row => {
          val xl = row.getString(0)
          val ddz = row.getString(4)
          !matchDDZ(xl, ddz, pdict)
        }).count()*/

    //println("incity=" + incityValue)
    //println("outcity=" + outcityValue)
    //println("outprovince=" + outprovinceValue)


    //InCityOrOut(ylcs, "incity.txt")
    spark.stop()

    def InCityOrOut(s: DataFrame, fileName: String): Unit = {
      val readerInCity = new FileReader("D:/projects/spark/src/main/resource/" + fileName)
      val bufferReader = new BufferedReader(readerInCity)
      val dict = bufferReader.readLine()
      bufferReader.close()
      readerInCity.close()
      // InCity市内，OutCity，
      val result = s.rdd.map(item => {
        val xl = item.getString(0)
        val ddz = item.getString(2)
        if (matchDDZ(xl, ddz, dict)) {
          ("InCity", 1)
        } else {
          ("OutCity", 1)
        }
      }).reduceByKey((x, y) => x + y)

      result.foreach(x => {
        println(x._1 + "," + x._2)
      })
    }

    def matchDDZ(xl: String, ddz: String, dict: String): Boolean = {
      val reg1 = "[0-9]"
      val reg2 = "（[\\u4e00-\\u9fa5]*）"
      val reg3 = "[(][0-9\\u4e00-\\u9fa5]*[)]"
      val reg4 = "服务站|下道车|机场|国际机场|车站|汽车站|长途车站|长途汽车站|中心站|中心|高速学生|高速|补|路口|总站|南站|北站|东站|西站|汽车"

      val xllen = xl.split("-").length
      if (dict.indexOf(ddz.replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
        true
      } else if (xllen == 2 && dict.indexOf(xl.split("-")(1).replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
        true
      } else if (xllen == 1 && dict.indexOf(xl.replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
        true
      } else if (xllen == 3 && dict.indexOf(xl.split("-")(2).replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
        true
      } else {
        false
      }
    }
  }
}
